package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2023/6/17
 * 查询维度的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName, Tuple2.of("id", id));
    }

    /**
     * 查询维度数据
     * 优化：旁路缓存
     * 思路：先从缓存中获取维度数据，如果缓存中存在要关联的维度，那么直接将其作为方法的返回值进行返回(缓存命中)；
     * 如果缓存中没有找到要关联的维度数据，那么发送请求到Phoenix表中进行查询，并将查询到的维度数据放到
     * 缓存中缓存起来，方便下次查询使用。
     * 缓存产品选型：
     * 状态：     性能高、维护性差
     * Redis：   性能也不错、维护性好     √
     * Redis相关的配置
     * key:        dim:维度表名:主键值1_主键值2
     * type：      string
     * expire:     1day
     * 注意：如果维度数据发生了变化，将缓存数据删除
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接从Redis中查询维度数据的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimJsonObj = null;

        try {
            // 先从缓存中获取维度数据
            jedis = RedisUtil.getJedis();
            dimInfoStr = jedis.get(redisKey.toString());

            if (StringUtils.isNotEmpty(dimInfoStr)) {
                //如果缓存中存在要关联的维度，那么直接将其作为方法的返回值进行返回(缓存命中)
                dimJsonObj = JSON.parseObject(dimInfoStr);
            } else {
                // 如果缓存中没有找到要关联的维度数据，那么发送请求到Phoenix表中进行查询
                System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);
                //底层还是调用PhoenixUtil的queryList方法
                List<JSONObject> dimList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
                if (dimList != null && dimList.size() > 0) {
                    //因为是根据维度的主键进行查询，所以集合中如果存在元素，只会有一条
                    dimJsonObj = dimList.get(0);
                    // 并将查询到的维度数据放到 缓存中缓存起来，方便下次查询使用。
                    if (jedis != null) {
                        jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObj.toJSONString());
                    }
                } else {
                    System.out.println("从Phoenix表中没有找到对应的维度数据~~");
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                System.out.println("~~~关闭Jedis客户端~~~");
                jedis.close();
            }
        }
        return dimJsonObj;
    }

    //查询维度数据--没有使用优化
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
            }
        }
        System.out.println("从Phoenix表中查询维度的SQL:" + selectSql);
        //底层还是调用PhoenixUtil的queryList方法
        List<JSONObject> dimList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if (dimList != null && dimList.size() > 0) {
            //因为是根据维度的主键进行查询，所以集合中如果存在元素，只会有一条
            dimJsonObj = dimList.get(0);
        } else {
            System.out.println("从Phoenix表中没有找到对应的维度数据~~");
        }
        return dimJsonObj;
    }

    //从Redis中将缓存的数据删除掉
    public static void deleteCached(String tableName, String id) {
        String redisKey = "dim:" + tableName.toLowerCase() + ":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(jedis != null){
                System.out.println("~~~删除缓存数据后，关闭Jedis~~~");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) {
        // System.out.println(getDimInfoNoCache("DIM_base_trademark", Tuple2.of("id", "1"),Tuple2.of("tm_name","xxx")));
        System.out.println(getDimInfo("DIM_base_trademark", "1"));
    }
}
